{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 并发模型概览"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现代并发程序中常见的协作模式旨在解决任务解耦、速率平衡、资源共享与扩展性等问题。下表概述几个核心模型及其适用场景：\n",
    "\n",
    "| 模型                   | 核心思路                                                  | 典型场景                         | 优势                       | 风险/注意事项                 | C++ 开源方案（示例）                                         |\n",
    "| ---------------------- | --------------------------------------------------------- | -------------------------------- | -------------------------- | ----------------------------- | ------------------------------------------------------------ |\n",
    "| **生产者-消费者**      | 通过缓冲区解耦生产与消费，平衡速率差异                    | 日志写入、任务分发、数据采集     | 异步解耦、吞吐提升         | 队列溢出 / 饥饿、防止虚假唤醒 | • **folly**：`folly::MPMCQueue`、`folly::ProducerConsumerQueue`<br>• **moodycamel::ConcurrentQueue**：高性能无锁队列<br>• **oneTBB**：`tbb::concurrent_bounded_queue` |\n",
    "| **发布-订阅**          | 基于主题或事件总线的松耦合通信                            | 消息系统、实时通知、事件驱动架构 | 动态扩展、广播能力强       | 顺序一致性、重复消费需控制    | • **ZeroMQ (cppzmq)**：轻量消息队列与 pub/sub<br>• **nngpp**：Nanomsg Next Gen 的 C++ 封装<br>• **CAF (Publisher API)**：Actor 框架内置 pub/sub |\n",
    "| **流水线（Pipeline）** | 多阶段串联，每阶段完成特定处理并传递结果                  | 流式数据处理、图像处理           | 易于阶段优化、并行度提升   | 阶段失衡导致瓶颈              | • **oneTBB Flow Graph**：节点和边构建流水线<br>• **HPX Pipelines**：异步任务链<br>• **Boost.Pipeline**：处理链条（实验性） |\n",
    "| **主从 / 工作队列**    | 中央调度器分配任务，工作线程处理                          | 任务分发、渲染农场、批量计算     | 调度集中化、可根据负载扩展 | 调度器成为瓶颈需容错          | • **Boost.Asio thread_pool**：任务提交 & 工作线程池<br>• **ctpl**：轻量线程池库<br>• **asio-grpc / agrpc**：任务队列驱动的 RPC |\n",
    "| **Actor**              | Actor 拥有私有状态，以消息异步通信                        | 分布式系统、游戏服务器           | 消除共享状态、便于横向扩展 | 消息调试复杂、顺序保证有限    | • **CAF (C++ Actor Framework)**：成熟的 Actor 平台<br>• **SObjectizer**：带 dispatchers 的 Actor/协程混合<br>• **Theron**：轻量 Actor 库 |\n",
    "| **CSP (Communicating Sequential Process) / Channel**      | 通过 Channel 传递数据并隐式同步                           | Go/Rust 风格协程、数据流编程     | 同步语义明确、易推理       | Channel 使用不当会死锁        | • **cppcoro**：`cppcoro::channel` 协程通道<br>• **libgo**：Go 风格协程和 channel<br>• **boost::fibers**：`buffered_channel`、`unbuffered_channel` |\n",
    "| **Fork–Join**          | 将任务拆分为子任务并行执行，最终汇总                      | 分治算法、并行排序               | 适合可拆分问题             | 大量短任务需配合工作窃取      | • **oneTBB parallel_invoke/parallel_for**：工作窃取调度器<br>• **Intel Threading Building Blocks**：任务调度 + dag<br>• **OpenMP**：`#pragma omp task` / `taskgroup` |\n",
    "| **Map–Reduce**         | Map 并行处理，Reduce 聚合结果                             | 大数据统计、日志分析             | 高扩展、容错成熟           | 编程模型受限、偏批处理        | • **Thrill**：分布式数据处理框架 (C++)<br>• **MR-MPI**：MPI 上的 MapReduce 库<br>• **Dispel4Py (C++ backends)**：支持 C++ 节点 |\n",
    "| **读者-写者**          | 读多写少场景，通过读写锁放宽读者并行                      | 配置缓存、共享资源               | 提升读取并发               | 写者饥饿需专门策略            | • **Boost.SharedMutex / Boost.UpgradeMutex**<br>• **folly::SharedMutex、`folly::RWSpinLock`**<br>• **pthread_rwlock (POSIX)** 的 C++ RAII 包装 |\n",
    "| **Reactor / Proactor** | 事件驱动 I/O，Reactor 分发就绪事件，Proactor 接收完成事件 | 高频网络服务、GUI 事件循环       | 单线程可处理大量 I/O       | 回调地狱、状态机复杂          | • **Boost.Asio**：Proactor 模型的异步 I/O<br>• **libuv**：跨平台 Reactor 循环<br>• **Seastar**：Future/Continuation 风格 Reactor |\n",
    "| **Barrier / Phaser**   | 多线程在关键点同步，等待全部到达后继续                    | 数值计算、阶段性流程             | 简单直观                   | 慢线程拖累整体，异常需处理    | • **Boost.Thread barrier/cyclic_barrier**<br>• **oneTBB barrier / latch**<br>• **HPX lcos::barrier / phased barrier** |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "模型选择要点\n",
    "\n",
    "1. **任务耦合度**：是否需要完全解耦生产与消费？若希望订阅者动态加入，倾向发布-订阅；若只是平衡速率，生产者-消费者即可。\n",
    "2. **状态共享策略**：Actor、CSP 倾向“消息+私有状态”；读写锁或生产者-消费者则在共享内存上同步。\n",
    "3. **负载特征**：流水线适合稳定阶段；Fork–Join 适合可拆分任务；工作队列适合任务颗粒均匀的批处理。\n",
    "4. **可扩展性需求**：消息驱动模型（Actor、Pub-Sub）更容易跨进程/机器扩展，传统锁模型更适合单机、低延迟场景。\n",
    "5. **调试与维护**：越抽象的模型越需成熟的监控/日志体系；简单模型（生产者-消费者、读者-写者）更易定位问题。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 生产者—消费者 （Producer–Consumer）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 解决目标       | 说明                                                         |\n",
    "| -------------- | ------------------------------------------------------------ |\n",
    "| 异步解耦       | 让生产者与消费者通过共享缓冲区（队列）交互，各自独立运行，消除直接调用带来的相互等待。 |\n",
    "| 平衡速率差异   | 缓冲区充当“蓄水池”，吸收生产/消费速度的短期不匹配，避免生产者被迫停工或消费者“饿死”。 |\n",
    "| 资源共享与互斥 | 明确采用锁、条件变量、信号量等机制保护共享队列，防止数据竞争和一致性问题。 |\n",
    "| 提高并发与吞吐 | 生产者与消费者可并行执行，多核环境下提升整体吞吐量与响应速度。 |\n",
    "| 模块化与可扩展 | 将数据生成与处理逻辑分离，便于独立扩展优化（如加减生产者/消费者），增强灵活性与可维护性。 |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 方案                         | C++ 标准 | 优点         | 适用场景             | 注意事项                   |\n",
    "| ---------------------------- | -------- | ------------ | -------------------- | -------------------------- |\n",
    "| `mutex + condition_variable` | C++11    | 简洁通用     | 常规队列             | 循环检查条件、避免通知丢失 |\n",
    "| `condition_variable_any`     | C++11    | 自定义锁协同 | 读写锁、复合锁       | 性能略低、锁逻辑更复杂     |\n",
    "| `counting_semaphore`         | C++20    | 精准资源计数 | 多生产/多消费        | 仍需互斥保护容器           |\n",
    "| `atomic + wait/notify`       | C++20    | 无锁低延迟   | 固定容量、高性能场景 | 内存序推理困难、易出错     |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 通知模型"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 仅生产者调用 notify\n",
    "\n",
    "std::queue<int> buffer;\n",
    "const unsigned int max_size = 10;\n",
    "std::mutex mtx;\n",
    "std::condition_variable cv;\n",
    "std::atomic<bool> done = false; // 用于通知消费者生产已经完成。一旦生产结束，消费者可以检测 done，并优雅退出。\n",
    "\n",
    "void producer() {\n",
    "    for (int i = 0; i < 20; ++i) { // 假设生产 20 个任务\n",
    "        std::unique_lock<std::mutex> lock(mtx);\n",
    "        cv.wait(lock, [] { return buffer.size() < max_size; });\n",
    "        buffer.push(i); // 生产\n",
    "        cv.notify_one();\n",
    "    }\n",
    "    done = true; // 生产完成\n",
    "    cv.notify_all(); // 通知消费者退出\n",
    "}\n",
    "\n",
    "void consumer() {\n",
    "    while (true) {\n",
    "        std::unique_lock<std::mutex> lock(mtx);\n",
    "        cv.wait(lock, [] { return !buffer.empty() || done; }); // 等待数据或退出\n",
    "        if (buffer.empty() && done) break; // 如果缓冲区为空且生产结束，退出\n",
    "        buffer.pop(); // 消费\n",
    "        cv.notify_one();\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 生产者与消费者都调用 notify\n",
    "\n",
    "std::queue<int> buffer;\n",
    "const unsigned int max_size = 10;\n",
    "std::mutex mtx;\n",
    "std::condition_variable producer_cv, consumer_cv;\n",
    "std::atomic<bool> done = false;\n",
    "\n",
    "void producer() {\n",
    "    for (int i = 0; i < 20; ++i) { // 假设生产 20 个任务\n",
    "        std::unique_lock<std::mutex> lock(mtx);\n",
    "        producer_cv.wait(lock, [] { return buffer.size() < max_size; });\n",
    "        buffer.push(i); // 生产\n",
    "        consumer_cv.notify_one();\n",
    "    }\n",
    "    done = true; // 生产完成\n",
    "    consumer_cv.notify_all(); // 通知消费者退出\n",
    "}\n",
    "\n",
    "void consumer() {\n",
    "    while (true) {\n",
    "        std::unique_lock<std::mutex> lock(mtx);\n",
    "        consumer_cv.wait(lock, [] { return !buffer.empty() || done; }); // 等待数据或退出\n",
    "        if (buffer.empty() && done) break; // 如果缓冲区为空且生产结束，退出\n",
    "        buffer.pop(); // 消费\n",
    "        producer_cv.notify_one();\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| **场景**                           | **是否需要 `cv.notify_one()`** | **原因**                                                     |\n",
    "| ---------------------------------- | ------------------------------ | ------------------------------------------------------------ |\n",
    "| 经典生产者-消费者模型              | 必须                           | 生产者和消费者通过条件变量相互通知，协调缓冲区的生产和消费   |\n",
    "| 生产者通知消费者      | 只需生产者调用                 | 消费者不需要通知生产者，因为生产者的逻辑不依赖消费者的状态   |\n",
    "| 简单的单向任务队列（无需条件变量） | 不需要                         | 生产者和消费者之间没有协调需求，生产者简单地生产固定任务，消费者逐一消费固定任务 |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `std::mutex + std::condition_variable`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[mutex_cv.cpp](./producer_consumer/mutex_cv.cpp), https://godbolt.org/z/91PhafGnq"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最经典的做法，利用互斥锁保护共享队列，条件变量负责等待/唤醒。\n",
    "- ✅ C++11 起即可使用，兼容性最好。\n",
    "- ✅ 适用绝大多数场景。\n",
    "- ⚠️ 需注意“循环检查条件”“通知丢失”等常见坑。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `std::condition_variable_any + std::shared_mutex`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[cv_any_shared_mutex.cpp](./producer_consumer/cv_any_shared_mutex.cpp), https://godbolt.org/z/jGWT5Y9W8"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当共享资源使用非 `std::mutex` 的锁（如读写锁、自旋锁）时，可换成 `condition_variable_any`。\n",
    "\n",
    "示例：利用 `std::shared_mutex` 实现多读单写场景，生产者仍使用独占写锁，消费者用读锁。\n",
    "\n",
    "- ✅ 支持任意满足 BasicLockable 的锁。\n",
    "- ⚠️ 性能较 `condition_variable` 略低，锁管理复杂度更高。\n",
    "- ⚠️ 读锁内仍需在弹出元素时升级为写锁，需谨慎。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `std::counting_semaphore + std::mutex`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[counting_semaphore.cpp](./producer_consumer/counting_semaphore.cpp), https://godbolt.org/z/W1EP79xcx"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "利用信号量控制“可消费”元素数量，减少条件变量的显式等待。\n",
    "- ✅ 信号量在高争用下比条件变量更轻量，支持准确的资源计数。\n",
    "- ✅ 多生产者/消费者扩展较自然。\n",
    "- ⚠️ 仍需互斥锁保护队列（除非采用无锁结构）。\n",
    "- ⚠️ C++20 才引入，对旧编译器不兼容。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `std::atomic + wait/notify`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[atomic_wait_notify.cpp](./producer_consumer/atomic_wait_notify.cpp), https://godbolt.org/z/PM8njxrEq"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当临界区极短且争用较小时，可使用原子变量实现轻量同步，减少锁开销。示例为固定大小环形缓冲区。\n",
    "\n",
    "- ✅ 无需互斥锁，适用于低延迟场景（如高性能队列、实时系统）。\n",
    "- ✅ `atomic::wait/notify` 提供内核级阻塞支持，避免忙等。\n",
    "- ⚠️ 实现复杂，对内存序、ABA 问题等要求高。\n",
    "- ⚠️ 固定缓冲区容量；若需动态队列需自定义无锁结构。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 发布-订阅"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[Eventbus](./event_bus/main.cpp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 文件                                  | 主要职责                                    | 关键点                                                       |\n",
    "| ------------------------------------- | ------------------------------------------- | ------------------------------------------------------------ |\n",
    "| `blocking_queue.h`                    | 提供线程安全的 `BlockingQueue<T>`           | 使用 `std::mutex` + `std::condition_variable` 实现多生产者/多消费者；`Stop()` 可让等待线程安全退出。 |\n",
    "| `message.h`                           | 定义消息结构 `Message`                      | 包含 `topic` 与 `payload` 两个字符串字段，作为事件载体。     |\n",
    "| `subscription.h` / `subscription.cpp` | 管理订阅生命周期的 RAII 封装 `Subscription` | 持有 `EventBus*`、主题、订阅 ID；析构或调用 `Reset()` 时自动 `Unsubscribe`；支持移动，不支持拷贝。 |\n",
    "| `event_bus.h` / `event_bus.cpp`       | 事件总线核心逻辑 `EventBus`                 | 提供 `Subscribe`、`Publish`、`Stop`；内部维护订阅者哈希表、阻塞队列、单独调度线程；使用 `std::shared_mutex` 协调读/写。 |\n",
    "| `main.cpp`                            | 演示程序                                    | 创建 `EventBus`、注册多个订阅者、启动发布线程、示范退订与停止流程。 |\n",
    "| `CMakeLists.txt`                      | 可选构建脚本                                | 设置 C++17、编译目标等。                                     |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![eventbus](../event_bus/EventBus.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 运行机制\n",
    "\n",
    "1. 订阅\n",
    "\n",
    "- 客户端调用 `EventBus::Subscribe(topic, callback)`。\n",
    "- 分配唯一订阅 ID，将 `{id, callback}` 存入 `subscribers_[topic]`。\n",
    "- 返回 `Subscription` 对象，客户端持有它来控制订阅生命周期。\n",
    "\n",
    "2. 发布\n",
    "\n",
    "- `Publish(topic, payload)` 将消息封装后推入 `BlockingQueue<Message>`。\n",
    "- 生产者线程立即返回，发布非阻塞。\n",
    "\n",
    "3. 调度线程\n",
    "\n",
    "- `EventBus` 构造时启动 `worker_` 线程，循环执行 `DispatchLoop()`。\n",
    "- 调度线程 `Pop()` 阻塞队列，取出消息。\n",
    "- 根据消息主题在 `subscribers_` 中查找匹配回调，并复制回调列表（持有共享锁，复制完成后释放锁）。\n",
    "- 依次执行回调，保证用户回调不持锁，避免阻塞其他发布/订阅操作。\n",
    "\n",
    "4. 退订 & 停止\n",
    "\n",
    "- `Subscription::Reset()` 或析构调用 `EventBus::Unsubscribe()`，从对应主题列表移除回调。\n",
    "- `EventBus::Stop()`：\n",
    "  1. 设置 `stopped_` 标志，调用 `queue_.Stop()` 唤醒阻塞等待；\n",
    "  2. `join` 工作线程；\n",
    "  3. 清空所有订阅者列表。\n",
    "- 再次调用 `Stop()` 无副作用（通过原子标志防止重复操作）。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 并发与同步策略\n",
    "\n",
    "- **阻塞队列**：`Push()` 与 `Pop()` 内部使用互斥锁，`Stop()` 设置停止标志并唤醒所有等待线程，确保弹出时返回 `std::nullopt`，安全退出循环。\n",
    "- **订阅表**：使用 `std::shared_mutex subs_mutex_`：\n",
    "  - `Subscribe` / `Unsubscribe` 获得独占锁，修改 `subscribers_`。\n",
    "  - `DispatchLoop` 在分发阶段获得共享锁，只复制回调列表。\n",
    "- **回调执行**：复制完回调向量后，在锁外执行，避免用户回调时间过长阻塞内部结构。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# CSP (Communicating Sequential Process) / Channel"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 文件           | 主要职责                      | 关键点                                                       |\n",
    "| -------------- | ----------------------------- | ------------------------------------------------------------ |\n",
    "| go/channel.h   | 实现 Go 语义的 `chan<T>` 模板 | 支持缓冲/无缓冲；多生产者/消费者；`send`/`recv`/`close`；使用 `std::mutex + std::condition_variable` 协调。 |\n",
    "| go/runtime.h   | 提供 `go::go(...)` 启动接口   | 封装 `std::thread`，模拟 Go 的 `go` 关键字；调用者需主动 `join`。 |\n",
    "| main.cpp       | 示例程序入口                  | 创建 `chan<std::string>`；启动生产者/消费者；演示 `close` 后资源回收。 |\n",
    "| CMakeLists.txt | 构建脚本                      | `aux_source_directory` 收集根目录与 `go` 目录源文件；设置 C++17、链接 pthread。 |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![csp](../channel/csp.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 运行机制\n",
    "\n",
    "1. **初始化**\n",
    "   - `go::make_chan<T>(capacity)` 创建通道，配置缓冲区容量、互斥锁与条件变量。\n",
    "   - `go::go(task, args...)` 启动生产者与消费者线程，返回 `std::thread` 对象以便后续 `join`。\n",
    "2. **发送（Producer）**\n",
    "   - 调用 `send(value)`：\n",
    "     - 无缓冲：等待接收方到达，将数据写入 `slot_`，完成握手后返回。\n",
    "     - 有缓冲：当缓冲区未满时推入队列；缓冲满时阻塞。\n",
    "   - 通道关闭（`closed_ == true`）时立即返回 `false`，用于终止发送循环。\n",
    "3. **接收（Consumer）**\n",
    "   - 调用 `recv(out)`：\n",
    "     - 无缓冲：等待 `slot_` 中出现数据，读取后唤醒发送方。\n",
    "     - 有缓冲：从队列取出数据；空队列时阻塞。\n",
    "   - 通道关闭且数据耗尽时返回 `false`，退出消费循环。\n",
    "4. **关闭与收尾**\n",
    "   - 主线程在适当时机调用 `close()`：\n",
    "     - 标记 `closed_ = true`，`notify_all` 唤醒所有阻塞的发送者/接收者。\n",
    "     - 剩余缓冲数据可继续被 `recv` 取走。\n",
    "   - `join` 所有线程，确保资源释放与日志收尾。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 并发与同步策略\n",
    "\n",
    "- **互斥与条件变量**：`mutex_` 保护缓冲区、状态位；`cv_send_`/`cv_recv_` 分别在缓冲区满/空时阻塞并唤醒对应线程。\n",
    "- **幂等关闭**：`close()` 内部检查已关闭状态，保证多次调用无副作用。\n",
    "- **引用语义**：禁用拷贝，只允许共享指针引用 `chan`，与 Go 的 channel 行为一致。\n",
    "- **线程治理**：通过 `go::go` 启动线程，但仍遵循 C++ 生命周期；示例中统一 `join` 防止悬挂线程。"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "name": "C++17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
